package com.glodon.paas.job.server.service.impl;

import static org.apache.http.params.HttpConnectionParams.*;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.params.HttpParams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;

import com.glodon.paas.job.model.Task;
import com.glodon.paas.job.server.service.NotifyService;

@Service
public class NotifyServiceImpl implements NotifyService {
    private static final Logger logger = LoggerFactory.getLogger(NotifyServiceImpl.class);

    private static final int DEFAULT_CONNECTION_TIMEOUT = 30000;
    private static final int DEFAULT_SOCKET_TIMEOUT = 30000;
    @Value("${job.server.connection.timeout}")
    private int connectionTimeout = DEFAULT_CONNECTION_TIMEOUT;
    @Value("${job.server.socket.timeout}")
    private int socketTimeout = DEFAULT_SOCKET_TIMEOUT;

    private ExecutorService threadPool = createDaemonThreadPool();
    @Value("${job.server.notify.retry.count}")
    private int notifyRetryCount;

    @Override
    public void notify(final Task task, final String clientUrl) {
        logger.debug("notify task, client url: {}, id: {}, type: {}",
                new Object[] { clientUrl, task.getTaskId(), task.getTaskType() });
        if (StringUtils.isBlank(clientUrl))
            return;
        final String json = task.toJson();
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < notifyRetryCount; i++) {
                    try {
                        HttpResponse resp = notifyJson(task.getTaskId(), json, clientUrl);
                        int statusCode = resp.getStatusLine().getStatusCode();
                        if (inStatus(statusCode, 200)) {
                            return;
                        }
                        if (inStatus(statusCode, 500)) {
                            logger.warn("notify client fail, reponse: {}", resp);
                        }
                    } catch (IOException e) {
                        logger.warn("notify client fail retry: {}", e.getMessage());
                    }
                }
                logger.warn("notify client fail, retry {}, client url: {}, id: {}, type: {}, body: {}", new Object[] {
                        clientUrl, notifyRetryCount, task.getTaskId(), task.getTaskType(), json });
            }

        });
    }

    private boolean inStatus(int statusCode, int in) {
        return statusCode >= in && statusCode < in + 100;
    }

    private HttpResponse notifyJson(String id, String json, String clientUrl) throws ClientProtocolException,
            IOException {
        HttpClient client = new DefaultHttpClient();
        HttpParams params = client.getParams();
        setSoTimeout(params, socketTimeout);
        setConnectionTimeout(params, connectionTimeout);
        String notifyUrl = clientUrl + (clientUrl.endsWith("/") ? "" : "/") + id;
        HttpPut put = new HttpPut(notifyUrl);
        HttpEntity entity = new StringEntity(json);
        logger.debug("notify client data : {}", json);
        put.setEntity(entity);
        HttpResponse resp = client.execute(put);
        logger.debug("notify json, url: {}, json: {}, status code: {}", notifyUrl, json, resp.getStatusLine()
                .getStatusCode());
        return resp;
    }

    private ExecutorService createDaemonThreadPool() {
        CustomizableThreadFactory threadFactory = new CustomizableThreadFactory("notify-thread-");
        threadFactory.setDaemon(true);
        return Executors.newCachedThreadPool(threadFactory);
    }
}
